package de.mmis.core.publishsubscribe;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.lang.ref.WeakReference;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InterfaceAddress;
import java.net.NetworkInterface;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;

import de.mmis.core.base.Pair;
import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.TreeParserException;
import de.mmis.core.base.abstracttree.serializer.DeserializableException;
import de.mmis.core.base.event.Event;
import de.mmis.core.config.Config;
import de.mmis.core.publishsubscribe.FilterHelper.ExtractResult;
import de.mmis.core.publishsubscribe.FilterManager.FilterEntry;
import de.mmis.core.publishsubscribe.PublishSubscribe.PublishSubscribeException;
import de.mmis.core.publishsubscribe.filter.Filter;
import de.mmis.core.sexpression.SExpressionParser;
import de.mmis.core.sexpression.SExpressionWriter;

/**
 * This class manages the connection between Publish/Subscribe Systems of other
 * hosts, and it's local clients.
 * 
 * @author Martin Nyolt
 * 
 */
class HostManager {

	private final static Logger LOGGER = Logger.getLogger(HostManager.class);

	private static class Client {
		Socket socket;
		BufferedInputStream input;
		BufferedOutputStream output;

		/**
		 * @throws IOException
		 *             when the streams could not be created
		 */
		Client(Socket socket) throws IOException {
			this.socket = socket;
			input = new BufferedInputStream(socket.getInputStream());
			output = new BufferedOutputStream(socket.getOutputStream());
		}

		@Override
		public String toString() {
			return socket.toString();
		}
	}

	private static final ThreadGroup localClientThreads = new ThreadGroup(
			"PubSub local client threads");

	private class LocalClientThread extends Thread {

		private Client client;

		public LocalClientThread(Client client) {
			super(localClientThreads, "PubSub client thread"
					+ client.socket.getRemoteSocketAddress());
			setDaemon(true);
			this.client = client;
		}

		@Override
		public void run() {
			// send hello message to client

			InnerNode hello = new InnerNode(new LeafNode(":hello"));
			hello.add(Tree.fromObject(remoteBindAddress.getHostAddress()));

			sendMessage(client, SExpressionWriter.writeSExpression(hello));

			while (running) {
				Tree expr = null;
				try {
					expr = SExpressionParser.parse(client.input, false);
				} catch (TreeParserException e) {
					if (!client.socket.isClosed()
							&& !e.unexpectedEndOfStreamOccurred())
						LOGGER.error("Can't parse input of client " + client, e);
					break;
				} catch (IOException e) {
					if (!client.socket.isClosed())
						LOGGER.error("Can't read input from client " + client,
								e);
					break;
				}

				if (client.socket.isClosed())
					break;

				if (expr == null)
					continue;

				Message message;
				try {
					message = expr.getAs(Message.class);
				} catch (DeserializableException e) {
					LOGGER.error("Can't deserialize message: " + expr, e);
					continue;
				}

				parseMessage(message, client);
			}

			try {
				client.socket.close();
			} catch (IOException e) {
				// ignore
			}

			removeClient(client);
		}
	}

	private class LocalListenThread extends Thread {

		public LocalListenThread() {
			super(LocalListenThread.class.getName());
			setDaemon(true);
		}

		@Override
		public void run() {
			while (running) {
				Socket clientSocket;
				try {
					clientSocket = localSocket.accept();
				} catch (IOException e) {
					if (running) {
						LOGGER.error("Can't accept new clients", e);
						// if we can't accept new clients the socket seems to be
						// broken
						running = false;
					}
					break;
				}

				try {
					Client client = new Client(clientSocket);
					synchronized (clients) {
						clients.add(client);
					}
					new LocalClientThread(client).start();
				} catch (IOException e) {
					LOGGER.error("Can't add client", e);
				}
			}

		}
	}

	private class RemoteReceiveThread extends Thread {

		public RemoteReceiveThread() {
			super("PubSub remote receive thread");
			setDaemon(true);
		}

		@Override
		public void run() {
			byte[] buf;
			try {
				// max UDP packet size is ~ 65000
				remoteSocket.setReceiveBufferSize(65000);
				buf = new byte[remoteSocket.getReceiveBufferSize()];
			} catch (SocketException e1) {
				buf = new byte[65000];
			}

			boolean wasError = false;

			while (running) {
				DatagramPacket packet = new DatagramPacket(buf, buf.length);
				try {
					remoteSocket.receive(packet);
					if (!running)
						break;
					parsePacket(packet);
					wasError = false;
				} catch (IOException e) {
					if (!running)
						// we wanted to get this exception
						break;

					LOGGER.warn("could not receive a packet", e);
					if (wasError) {
						LOGGER.error("Two errors in a line - try to close the socket and create a new one");
						remoteSocket.close();
						try {
							initRemote();
						} catch (PublishSubscribeException e1) {
							LOGGER.error("Unable to create a new socket - giving up");
							HostManager.this.stop();
							break;
						}
						wasError = false;
					} else
						wasError = true;
				}
			}
		}
	}

	private volatile boolean running = false;

	private ServerSocket localSocket;
	protected LinkedList<Client> clients = new LinkedList<Client>();

	private InetAddress remoteBindAddress;
	private DatagramSocket remoteSocket;
	private DatagramSocket remoteSendSocket;
	private InetAddress remoteBroadcastAddress;

	/**
	 * {@link Filter}s local clients have subscribed. Must be checked against
	 * incoming {@link Event}s
	 * 
	 * @see #inputFilterMap
	 */
	private final FilterManager<Client, Integer> inputFilters = new FilterManager<Client, Integer>();

	/**
	 * Maps local client ids to the ids of {@link #inputFilters}
	 */
	private final HashMap<Pair<Client, Integer>, Integer> inputFilterMap = new HashMap<Pair<Client, Integer>, Integer>();

	private final HashMap<String, Map<String, Tree>> worldstate = new HashMap<String, Map<String, Tree>>();

	// /**
	// * {@link Filter}s remote hosts have subscribed. Must be checked against
	// * outgoing {@link Event}s
	// *
	// * @see #outputFilterMap
	// */
	// private final FilterManager<InetAddress, Integer> outputFilters = new
	// FilterManager<InetAddress, Integer>();
	//
	// /**
	// * Maps remote client ids to the ids of {@link #outputFilters}
	// */
	// private final HashMap<Pair<InetAddress, Integer>, Integer>
	// outputFilterMap = new HashMap<Pair<InetAddress, Integer>, Integer>();

	// Using a WeakReference here to allow the GC to finalize the instance
	// if no one uses it. If the HostManager is running, the member classes
	// (LocalListenThred etc.) will still have a reference, thus preventing
	// a collection
	private static WeakReference<HostManager> weakInstance;

	static HostManager getInstance() {
		return getInstance(null);
	}

	static HostManager getInstance(InetAddress bindAddress) {
		HostManager instance = null;

		if (weakInstance != null)
			instance = weakInstance.get();

		if (instance == null) {
			instance = new HostManager(bindAddress);
			weakInstance = new WeakReference<HostManager>(instance);
		}

		return instance;
	}

	private HostManager(InetAddress bindAddress) {
		// singleton, hide constructor
		this.remoteBindAddress = bindAddress;
	}

	void init() throws PublishSubscribeException {
		if (running) {
			LOGGER.debug("ignoring init, already running");
			return;
		}

		LOGGER.trace("attempting to create sockets");

		try {
			localSocket = new ServerSocket(Config.PUBSUB_PORT);
		} catch (IOException e) {
			throw new PublishSubscribeException("Can't create local socket", e);
		}

		try {
			initRemote();
		} catch (PublishSubscribeException e) {
			try {
				localSocket.close();
			} catch (IOException e1) {
				// ignore
			}
			throw new PublishSubscribeException("Can't create remote socket", e);
		}

		running = true;
		new LocalListenThread().start();
		new RemoteReceiveThread().start();

		LOGGER.trace("sockets created");
	}

	/**
	 * creates the {@link #remoteSocket} and {@link #remoteSendSocket} and inits
	 * {@link #remoteBindAddress} and {@link #remoteBroadcastAddress}
	 * 
	 * @throws PublishSubscribeException
	 */
	void initRemote() throws PublishSubscribeException {
		LOGGER.trace("creating remote socket");

		if (remoteBindAddress == null || remoteBroadcastAddress == null) {
			Enumeration<NetworkInterface> nis;
			try {
				nis = NetworkInterface.getNetworkInterfaces();
			} catch (SocketException e1) {
				throw new PublishSubscribeException(
						"Could not enumerate all network interfaces", e1);
			}
			ifLoop: while (nis.hasMoreElements()) {
				NetworkInterface ni = nis.nextElement();
				try {
					if (ni.isLoopback())
						continue;
				} catch (SocketException e) {
					LOGGER.warn("Could not check if " + ni + " is loopback", e);
					continue;
				}

				List<InterfaceAddress> as = ni.getInterfaceAddresses();
				for (InterfaceAddress a : as) {
					if (a.getBroadcast() != null) {
						remoteBindAddress = a.getAddress();
						remoteBroadcastAddress = a.getBroadcast();
						if (remoteBindAddress.getHostAddress().startsWith(
								"139.30."))
							break ifLoop;
					}
				}
			}
		}

		if (remoteBindAddress == null)
			throw new PublishSubscribeException(
					"No none-loopback network interface found");

		// always set broadcast to 255.255.255.255
		try {
			remoteBroadcastAddress = InetAddress.getByName("255.255.255.255");
		} catch (UnknownHostException e) {
			// should not happen
			LOGGER.error("Could not get InetAddress of 255.255.255.255", e);
		}

		if (remoteBroadcastAddress == null)
			throw new PublishSubscribeException("Have no broadcast address");

		LOGGER.info("bindAddress: " + remoteBindAddress + " - broadcast: "
				+ remoteBroadcastAddress);

		try {
			remoteSocket = new DatagramSocket(Config.PUBSUB_PORT);
			remoteSendSocket = new DatagramSocket(0, remoteBindAddress);
		} catch (IOException e) {
			throw new PublishSubscribeException("Can't create remote socket", e);
		}
	}

	void stop() {
		if (!running) {
			LOGGER.debug("ignoring stop, already stopped");
			return;
		}

		LOGGER.debug("stopping, closing sockets");

		running = false;

		try {
			localSocket.close();
		} catch (IOException e1) {
			// ignore
		}
		remoteSocket.close();
		remoteSendSocket.close();

		synchronized (clients) {
			for (Client client : clients) {
				try {
					client.socket.close();
				} catch (IOException e) {
					// ignore
				}
			}

			if (!clients.isEmpty())
				// wait until clients is empty
				try {
					clients.wait();
				} catch (InterruptedException e) {
					LOGGER.warn("Could not wait until clients is empty", e);
				}
		}

		LOGGER.debug("stopped");
	}

	/**
	 * parses a message from a local client
	 */
	private void parseMessage(Message message, Client client) {
		LOGGER.trace("Parsing message of local client " + client + ": "
				+ message);

		Message m;
		Integer id;

		switch (message.getType()) {
		case EVENT:
			processEvent(message.getEvent(), message.getDeliver(), client);
			break;

		case SUBSCRIBE:
			id = inputFilters.addFilter(message.getClazz(),
					message.getFilter(), client, message.getId());

			synchronized (inputFilterMap) {
				inputFilterMap.put(
						new Pair<Client, Integer>(client, message.getId()), id);
			}

			// handle extracted worldstate filters
			ExtractResult result = FilterHelper.extractWorldstateFilter(message
					.getFilter());
			synchronized (worldstate) {
				for (Pair<String, String> entry : result.worldstateFilter
						.values()) {
					// if the worldstate does not contain this key, add "null"
					// if we receive any event, we check in the worldstate-Map
					// if we
					// are interested in the value

					String deviceID = entry.getFirst();
					String key = entry.getSecond();

					Map<String, Tree> ed = worldstate.get(deviceID);
					if (ed == null) {
						LOGGER.debug("Adding device " + deviceID
								+ " to the worldstate");
						ed = new HashMap<String, Tree>();
						worldstate.put(deviceID, ed);

						// the client requests the current state
					}
					if (!ed.containsKey(key)) {
						LOGGER.debug("Adding key " + key + " of device "
								+ deviceID + " to the worldstate");
						ed.put(key, null);
					}
				}
			}
			break;

		case UNSUBSCRIBE:
			LinkedList<Integer> ids = new LinkedList<Integer>();

			synchronized (inputFilterMap) {
				for (Integer i : message.getIds())
					ids.add(inputFilterMap.remove(new Pair<Client, Integer>(
							client, i)));
			}

			for (Integer i : ids)
				inputFilters.removeFilter(i);

			rebuildWorldstate();
			break;

		case STATE_REQUEST:
			m = Message.createStateRequest(message.getDeviceID(),
					message.getKey(), null);
			sendBroadcastMessage(m);
			break;

		case STATE_ANSWER:
			synchronized (worldstate) {
				Map<String, Tree> state = worldstate.get(message.getDeviceID());
				if (state != null && state.containsKey(message.getKey()))
					state.put(message.getKey(), message.getValue());
			}

			InetAddress dest;
			try {
				dest = InetAddress.getByName(message.getDestIP());
			} catch (UnknownHostException e) {
				LOGGER.error(
						"Can't create InetAddress from IP "
								+ message.getDestIP(), e);
				break;
			}
			m = Message.createStateAnswer(message.getDeviceID(),
					message.getKey(), message.getValue(), null);
			sendUnicastMessage(m, dest);
			break;
		}
	}

	private void processEvent(PlainEvent event, boolean deliver, Client client) {
		// received event from local client
		Message m = Message.createEvent(event, deliver);

		synchronized (worldstate) {
			// update worldstate
			Map<String, Tree> eventData = event.eventData;
			Tree sender = eventData.get(":sender");
			if (sender != null) {
				String id;
				try {
					id = sender.getAsString();
					Map<String, Tree> state = worldstate.get(id);
					if (state != null)
						for (String key : eventData.keySet())
							if (state.containsKey(key)) {
								LOGGER.trace("Updating key " + key
										+ " of device " + sender
										+ " in the worldstate");
								state.put(key, eventData.get(key));
							}
				} catch (DeserializableException e) {
					LOGGER.error(sender + " is not a string", e);
				}
			}

			// check for other local VMs interested in the Event
			// no worldstate here
			for (Client c : inputFilters.getForEvent(event, worldstate)
					.keySet())
				if (c != client)
					sendMessage(c, m);
		}

		if (client != null)
			// only broadcast when the event was received from a local client
			sendBroadcastMessage(m);
	}

	private void rebuildWorldstate() {
		LOGGER.trace("rebuilding worldstate");

		// reconstruct all worldstateFilters and check if we don't need some
		HashMap<Filter, Pair<String, String>> filters = new HashMap<Filter, Pair<String, String>>();

		for (FilterEntry<Client, Integer> filterEntry : inputFilters
				.getAllEntries()) {
			HashMap<Filter, Pair<String, String>> fs;
			fs = FilterHelper.extractWorldstateFilter(filterEntry.getFilter()).worldstateFilter;
			filters.putAll(fs);
		}

		// remove entries from worldstate
		synchronized (worldstate) {
			Collection<Pair<String, String>> values = filters.values();

			Iterator<Entry<String, Map<String, Tree>>> it = worldstate
					.entrySet().iterator();
			while (it.hasNext()) {
				// for every ID - KEY entry in the worldstate,
				// check if this pair is in the values collection,
				// i. e. if it's still needed
				Entry<String, Map<String, Tree>> entry = it.next();
				String id = entry.getKey();
				Map<String, Tree> state = entry.getValue();

				Iterator<String> itState = state.keySet().iterator();
				while (itState.hasNext()) {
					String key = itState.next();
					Pair<String, String> pair = new Pair<String, String>(id,
							key);
					if (values.contains(pair))
						// this key is still needed
						continue;

					// this key is not needed any longer
					LOGGER.trace("removing key " + key + " of device " + id
							+ " from the worldstate");
					itState.remove();
				}

				if (state.isEmpty()) {
					// the hole device is not needed any longer
					LOGGER.trace("remove device " + id + " from the worldstate");
					it.remove();
				}
			}
		}
	}

	/**
	 * parses a remote message
	 */
	private void parsePacket(DatagramPacket packet) {
		// avoid parsing packet send after calling stop()
		if (!running)
			return;

		byte[] b = Arrays.copyOf(packet.getData(), packet.getLength());
		Tree expr;
		try {
			expr = SExpressionParser.parse(b);
		} catch (TreeParserException e) {
			LOGGER.error("Could not parse S-Expression of packet " + b, e);
			return;
		}

		Message message;
		try {
			message = expr.getAs(Message.class);
		} catch (DeserializableException e) {
			LOGGER.error("Could not deserialize message of expression " + expr,
					e);
			return;
		}

		InetAddress address = packet.getAddress();
		if (!remoteBindAddress.equals(address))
			parseMessage(message, address);
	}

	/**
	 * parses a remote message
	 */
	private void parseMessage(Message message, InetAddress address) {
		LOGGER.trace("Parsing message from remote " + address + ": " + message);

		// Integer id;
		Message m;

		switch (message.getType()) {
		case EVENT:
			processEvent(message.getEvent(), message.getDeliver(), null);

			break;

		case SUBSCRIBE:
			// id = outputFilters.addFilter(message.getClazz(),
			// message.getFilter(), address, message.getId());
			//
			// synchronized (outputFilterMap) {
			// outputFilterMap.put(new Pair<InetAddress, Integer>(address,
			// message.getId()), id);
			// }
			//
			// m = Message.createSubscribe(message.getFilter(),
			// message.getClazz(), id, true);
			// sendMessage(m);
			//
			// break;
		case UNSUBSCRIBE:
			// LinkedList<Integer> ids = new LinkedList<Integer>();
			// synchronized (outputFilterMap) {
			// for (Integer i : message.getIds())
			// ids.add(outputFilterMap.get(new Pair<InetAddress, Integer>(
			// address, i)));
			// }
			//
			// for (Integer i : ids)
			// outputFilters.removeFilter(i);
			//
			// m = Message.createUnsubscribe(ids, true);
			// sendMessage(m);
			// break;
			LOGGER.warn("remote un/subscribe not supported");
			break;

		case STATE_REQUEST:
			String ip = address.getHostAddress();
			m = Message.createStateRequest(message.getDeviceID(),
					message.getKey(), ip);
			sendMessage(m);
			break;

		case STATE_ANSWER:
			synchronized (worldstate) {
				Map<String, Tree> state = worldstate.get(message.getDeviceID());
				if (state != null && state.containsKey(message.getKey()))
					state.put(message.getKey(), message.getValue());
			}

			m = Message.createStateAnswer(message.getDeviceID(),
					message.getKey(), message.getValue(), null);
			sendMessage(m);
			break;
		}
	}

	/**
	 * removed the client, cancels all subscribtions
	 * 
	 * @param client
	 */
	private void removeClient(Client client) {
		LinkedList<Integer> ids = new LinkedList<Integer>();

		synchronized (inputFilterMap) {
			for (FilterEntry<Client, Integer> entry : inputFilters
					.removeClient(client)) {
				ids.add(inputFilterMap.remove(new Pair<Client, Integer>(client,
						entry.getID())));
			}
		}

		synchronized (clients) {
			clients.remove(client);
			if (clients.isEmpty())
				clients.notifyAll();
		}

		rebuildWorldstate();
	}

	/**
	 * broadcasts the message via UDP to other HostManagers.
	 */
	private void sendBroadcastMessage(Message message) {
		Tree e = Tree.fromObject(message);
		LOGGER.trace("Remote: broadcasting " + message);
		byte[] b = SExpressionWriter.writeSExpression(e);

		DatagramPacket packet = new DatagramPacket(b, b.length,
				remoteBroadcastAddress, Config.PUBSUB_PORT);

		try {
			remoteSendSocket.send(packet);
		} catch (IOException ex) {
			LOGGER.error("Sending UDP broadcast failed, Message: " + message
					+ " - Tree: " + e, ex);
		}
	}

	/**
	 * sends the message via UDP to one other HostManager.
	 */
	private void sendUnicastMessage(Message message, InetAddress address) {
		Tree e = Tree.fromObject(message);
		LOGGER.trace("Remote: sending to " + address + ": " + message);
		byte[] b = SExpressionWriter.writeSExpression(e);

		DatagramPacket packet = new DatagramPacket(b, b.length, address,
				Config.PUBSUB_PORT);

		try {
			remoteSendSocket.send(packet);
		} catch (IOException ex) {
			LOGGER.error("Sending UDP broadcast failed", ex);
		}

	}

	/**
	 * Sends the message to all local clients
	 */
	private void sendMessage(Message message) {
		Tree e = Tree.fromObject(message);
		LOGGER.trace("Sending to all clients: " + message);
		byte[] b = SExpressionWriter.writeSExpression(e);
		for (Client c : clients)
			sendMessage(c, b);
	}

	/**
	 * Sends the message to the client
	 */
	private void sendMessage(Client client, Message message) {
		Tree e = Tree.fromObject(message);
		LOGGER.trace("Sending to client " + client + ": " + message);
		byte[] b = SExpressionWriter.writeSExpression(e);
		sendMessage(client, b);
	}

	/**
	 * Sends the message to the client
	 */
	private void sendMessage(Client client, byte[] b) {
		synchronized (client) {
			try {
				client.output.write(b);
				client.output.flush();
			} catch (IOException e) {
				LOGGER.error(
						"Could not send message to local client " + client, e);
				try {
					client.socket.close();
				} catch (IOException e1) {
					// ignore
				}
			}
		}
	}
}
